package org.apache.flink.formats.jsontm;

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.formats.common.TimestampFormat;
import org.apache.flink.formats.json.JsonFormatOptionsUtil;
import org.apache.flink.formats.json.JsonToRowDataConverters;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.json.JsonReadFeature;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationFeature;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.DecimalType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
import org.apache.flink.util.Collector;

import java.io.IOException;
import java.util.List;
import java.util.Objects;

import static org.apache.flink.formats.json.JsonFormatOptions.FAIL_ON_MISSING_FIELD;
import static org.apache.flink.formats.json.JsonFormatOptions.IGNORE_PARSE_ERRORS;
import static org.apache.flink.formats.jsontm.JsonTmFormatFactory.ROOT_PATH;

/**
 * <ol>
 * date:2022/1/5 editor:yanghongjian
 * <li>创建文档</li>
 * <li>运行时逻辑:用于将 Bytes 转换为具有 Integer、String等Row类型，并带有 RowKind 信息</li>
 * </ol>
 * <ol>
 *
 * @author <a href="mailto:12719889@qq.com">YangHongJian</a>
 * @version 3.0
 * @since 1.8
 */
public class JsonTmDeserializationSchema implements DeserializationSchema<RowData> {
    private final ReadableConfig options;
    //physical + meta data
    private final TypeInformation<RowData> producedTypeInfo;
    private final ObjectMapper objectMapper = new ObjectMapper();
    private final JsonToRowDataConverters.JsonToRowDataConverter[] metadataConverters;
    private final JsonTmPathManager jsonTmPathManager;

    public JsonTmDeserializationSchema(
            ReadableConfig options, DataType physicalDataType,
            List<JsonTmDecodingFormat.ReadableMetadata> requestedMetadata,
            TypeInformation<RowData> producedTypeInfo
    ) {
        this.options = options;
        final RowType jsonRowType = (RowType) physicalDataType.getLogicalType();

        TimestampFormat timestampFormat = JsonFormatOptionsUtil.getTimestampFormat(this.options);
        String rootPath = options.get(ROOT_PATH);
        boolean ignoreParseErrors = options.get(IGNORE_PARSE_ERRORS);
        boolean failOnMissingField = options.get(FAIL_ON_MISSING_FIELD);

        JsonToRowDataConverters dataConverters = new JsonToRowDataConverters(failOnMissingField, ignoreParseErrors, timestampFormat);
        if (LogicalTypeChecks.hasNested(jsonRowType, (t) -> t instanceof DecimalType)) {
            this.objectMapper.enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS);
        }
        this.metadataConverters = new JsonToRowDataConverters.JsonToRowDataConverter[jsonRowType.getFields().size() + requestedMetadata.size()];
        String[] paths = new String[jsonRowType.getFields().size() + requestedMetadata.size()];
        List<RowType.RowField> fields = jsonRowType.getFields();
        final int physicalArity = fields.size();
        final int metadataArity = requestedMetadata.size();
        for (int physicalPos = 0; physicalPos < fields.size(); physicalPos++) {
            RowType.RowField field = fields.get(physicalPos);
            this.metadataConverters[physicalPos] = dataConverters.createConverter(field.getType());
            paths[physicalPos] = rootPath.isEmpty() ? field.getName() : (rootPath + "." + field.getName());
        }
        for (int metadataPos = 0; metadataPos < metadataArity; metadataPos++) {
            JsonTmDecodingFormat.ReadableMetadata readableMetadata = requestedMetadata.get(metadataPos);
            this.metadataConverters[physicalArity + metadataPos] = dataConverters.createConverter(readableMetadata.dataType.getLogicalType());
            paths[physicalArity + metadataPos] = readableMetadata.key;
        }

        this.objectMapper.configure(JsonReadFeature.ALLOW_UNESCAPED_CONTROL_CHARS.mappedFeature(), true);

        this.jsonTmPathManager = new JsonTmPathManager(paths);
        this.producedTypeInfo = producedTypeInfo;
    }

    @Override
    public RowData deserialize(byte[] message) {
        throw new RuntimeException(
                "Please invoke DeserializationSchema#deserialize(byte[], Collector<RowData>) instead.");
    }

    @Override
    public void deserialize(byte[] message, Collector<RowData> out) throws IOException {
        if (message == null || message.length == 0) {
            return;
        }
        final JsonNode rootNode = this.objectMapper.readTree(message);
        jsonTmPathManager.consumer(rootNode, values-> emitRow(values, out));
    }

    private void emitRow(JsonNode[] values, Collector<RowData> out) {
        final int fieldSize = metadataConverters.length;
        final GenericRowData producedRow = new GenericRowData(fieldSize);
        for (int pos = 0; pos < fieldSize; pos++) {
            producedRow.setField(pos, metadataConverters[pos].convert(values[pos]));
        }
        out.collect(producedRow);
    }

    @Override
    public boolean isEndOfStream(RowData nextElement) {
        return false;
    }

    @Override
    public TypeInformation<RowData> getProducedType() {
        return producedTypeInfo;
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj) {
            return true;
        }
        if (obj == null || getClass() != obj.getClass()) {
            return false;
        }
        JsonTmDeserializationSchema that = (JsonTmDeserializationSchema) obj;
        return Objects.equals(producedTypeInfo, that.producedTypeInfo);
    }

    @Override
    public int hashCode() {
        return Objects.hash(producedTypeInfo);
    }

}
